package com.kasax.httpclient.core;

import com.kasax.httpclient.utils.LogUtils;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class Dispatcher {
    // 最大請求數
    private final int maxRequest;
    // 同一主机最大请求数
    private final int maxRequestPreHost;

    private ExecutorService executorService;
    private AtomicInteger threadNum = new AtomicInteger();

    // 执行中的队列 正在执行队列
    private Deque<RealCall.AsyncCall> runningAsyncCalls = new ArrayDeque<>();
    //  等待中的队列-异步  等待执行队列
    private Deque<RealCall.AsyncCall> readAsyncCalls = new ArrayDeque<>();
    //  执行中的队列-同步
    private Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

    public Dispatcher() {
        this(64, 10);
    }


    public Dispatcher(int maxRequest, int maxRequestPreHost) {
        this.maxRequest = maxRequest;
        this.maxRequestPreHost = maxRequestPreHost;
    }

    private synchronized ExecutorService executorService() {
        if (executorService == null) {
            ThreadFactory threadFactory = new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r, "CoreThread-" + threadNum.incrementAndGet());
                    return thread;
                }
            };
            /**
             *    1、corePoolSize：线程池中核心线程数的最大值
             *    2、maximumPoolSize：线程池中能拥有最多线程数
             *    3、keepAliveTime：表示空闲线程的存活时间  60秒
             *    4、表示keepAliveTime的单位。
             *    5、workQueue：它决定了缓存任务的排队策略。
             *      SynchronousQueue<Runnable>：此队列中不缓存任何一个任务。向线程池提交任务时，
             *      如果没有空闲线程来运行任务，则入列操作会阻塞。当有线程来获取任务时，
             *      出列操作会唤醒执行入列操作的线程。
             *    6、指定创建线程的工厂
             */
            executorService = new ThreadPoolExecutor(
                    0,
                    Integer.MAX_VALUE,
                    60,
                    TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>(),
                    threadFactory
            );
        }
        return executorService;
    }

    public void enqueue(RealCall.AsyncCall asyncCall) {
        LogUtils.i(getClass(), "enqueue");
        if (runningAsyncCalls.size() < maxRequest && runningSyncCallsForHost(asyncCall) < maxRequestPreHost) {
            LogUtils.i(getClass(), "enqueue 提交执行");
            runningAsyncCalls.add(asyncCall);
            executorService().execute(asyncCall);
        } else {
            LogUtils.i(getClass(), "enqueue 等待执行");
            readAsyncCalls.add(asyncCall);
        }
    }

    public void execute(RealCall call) {
        runningSyncCalls.add(call);
    }

    public void finish(RealCall call) {
        finished(runningSyncCalls, call, false);
    }


    public void finish(RealCall.AsyncCall asyncCall) {
        finished(runningAsyncCalls, asyncCall, true);
    }

    /*
     *请求结束 移出正在运行队列
     *并判断是否执行等待队列中的请求
     */
    @Deprecated
    public void finished(RealCall.AsyncCall asyncCall) {
        synchronized (this) {
            runningAsyncCalls.remove(asyncCall);
            //判断是否执行等待队列中的请求
            promoteCalls();
        }
    }

    private <T> void finished(Deque<T> calls, T call, boolean promoteCalls) {
        synchronized (this) {
            if (!calls.remove(call)) {
                throw new AssertionError("Call wasn't in -flight!");
            }
            if (promoteCalls) {
                promoteCalls();
            }
        }
    }

    // 掃描等待执行任务队列，将任务放入正在执行任务队列，并执行该任务
    // 判断是否执行等待队列中的请求
    private void promoteCalls() {
        //同时请求达到上限
        if (runningAsyncCalls.size() >= maxRequest) {
            return;
        }
        //没有等待执行请求
        if (readAsyncCalls.isEmpty()) {
            return;
        }
        for (Iterator<RealCall.AsyncCall> iterator = readAsyncCalls.iterator(); iterator.hasNext(); ) {
            RealCall.AsyncCall call = iterator.next();
            //同一host同时请求为达上限
            if (runningSyncCallsForHost(call) < maxRequest) {
                // 从等待队列中移除该任务
                iterator.remove();
                // 添加到执行的队列中
                runningAsyncCalls.add(call);
                // 执行该任务
                executorService().execute(call);
            }
            //到达同时请求上限
            if (runningAsyncCalls.size() >= maxRequest) {
                return;
            }
        }
    }

    /**
     * 同一host的同时请求数据
     *
     * @param call
     * @return
     */
    private int runningSyncCallsForHost(RealCall.AsyncCall call) {
        int result = 0;
        for (RealCall.AsyncCall asyncCall : runningAsyncCalls) {
            if (asyncCall.host().equals(call.host())) {
                result++;
            }
        }
        return result;
    }
}
